package com.ctrip.corp.obt.shard.service;

import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

import javax.annotation.PostConstruct;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource;
import org.springframework.stereotype.Component;

import com.ctrip.corp.obt.generic.core.context.TenantContext;
import com.ctrip.corp.obt.generic.utils.CollectionUtils;
import com.ctrip.corp.obt.generic.utils.JsonUtils;
import com.ctrip.corp.obt.shard.utils.SecurityFuzzyQueryUtils;
import com.ctrip.corp.obt.shard.utils.SecurityFuzzyQueryUtils.DividingColumnInfo;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

/**
 * 历史数据的分词列信息的同步
 *
 * @author xukang zhang
 * @date 2024/1/9 12:52
 */
@Component
@Slf4j
public class SyncHistoryDivideColumnService {
    /**
     * 批量查询的数量
     */
    public static final int BATCH_NUMBER = 100;

    @Autowired
    private AbstractRoutingDataSource routingDataSource;
    private JdbcTemplate jdbcTemplate;

    @PostConstruct
    public void init() {
        jdbcTemplate = new JdbcTemplate(routingDataSource);
    }

    /**
     * 同步分词列
     *
     * @return
     */
    public <T> SyncHistoryDivideColumnResponse<?>
        doSync(SyncHistoryDivideColumnRequest<T> syncHistoryDivideColumnRequest) {
        // 设置租户id
        String tenantId = syncHistoryDivideColumnRequest.getTenantId();
        TenantContext.setTenantId(tenantId);
        return syncHistoryDivideColumnGeneric(syncHistoryDivideColumnRequest);
    }

    /**
     * 查询历史数据，并根据entity类中的SecurityFuzzyQueryField注解的配置，对分词列进行分词加密后，更新进表中分词列
     */
    private <T> SyncHistoryDivideColumnResponse<?>
        syncHistoryDivideColumnGeneric(SyncHistoryDivideColumnRequest<T> syncHistoryDivideColumnRequest) {
        // 1. 查询需要同步的数据
        String boundaryId = null;
        String conditionalSql = MessageFormat.format("select * from {0} where {1} > ? order by {1} limit {2}",
            syncHistoryDivideColumnRequest.getTableName(), syncHistoryDivideColumnRequest.getIdColumnName(),
            BATCH_NUMBER);
        String ordinarySql = MessageFormat.format("select * from {0} order by {1} limit {2}",
            syncHistoryDivideColumnRequest.getTableName(), syncHistoryDivideColumnRequest.getIdColumnName(),
            BATCH_NUMBER);

        List<T> missingDataList = CollectionUtils.newArrayList();
        long totalCount = 0;
        long updateCount = 0;

        while (true) {
            List<T> baseDataList = null;
            if (boundaryId == null) {
                baseDataList = jdbcTemplate.query(ordinarySql,
                    BeanPropertyRowMapper.newInstance(syncHistoryDivideColumnRequest.getEntityClazz()));
            } else {
                baseDataList = jdbcTemplate.query(conditionalSql,
                    BeanPropertyRowMapper.newInstance(syncHistoryDivideColumnRequest.getEntityClazz()), boundaryId);
            }
            if (CollectionUtils.isEmpty(baseDataList)) {
                // 没有未同步的数据，则返回
                break;
            }
            totalCount += baseDataList.size();
            // 取最后一条数据的id作为下一次查询的边界条件
            T endRecord = baseDataList.get(baseDataList.size() - 1);
            boundaryId = syncHistoryDivideColumnRequest.getIdGetter().apply(endRecord);
            if (boundaryId == null) {
                log.error("End sync while getting boundaryId failed, data: " + JsonUtils.toJsonString(endRecord));
                break;
            }

            // 同步这一批数据
            for (T baseData : baseDataList) {
                // 2. 对需要同步的数据进行分词加密
                List<DividingColumnInfo> dividingColumnInfoList =
                    SecurityFuzzyQueryUtils.encryptWithDividingByEntity(baseData,
                        syncHistoryDivideColumnRequest.getTableName(),
                        syncHistoryDivideColumnRequest.getDecryptFunction());
                // 3. 更新到历史表
                int count = 0;
                try {
                    count = updateTables(dividingColumnInfoList, syncHistoryDivideColumnRequest.getIdColumnName(),
                        syncHistoryDivideColumnRequest.getIdGetter().apply(baseData));
                } catch (Exception e) {
                    log.error("Save Divide Column failed, data: " + JsonUtils.toJsonString(baseData), e);
                }
                if (count == 0) {
                    missingDataList.add(baseData);
                } else {
                    updateCount++;
                }
            }
        }

        log.info("Security Fuzzy Sync End, count: {}, total: {}, missingData: {}", updateCount,
            totalCount, JsonUtils.toJsonString(missingDataList));
        if (updateCount < totalCount) {
            return SyncHistoryDivideColumnResponse.fail(missingDataList,
                "总数据: " + totalCount + ", 已同步数据: " + updateCount + ", 部分数据同步失败");
        }
        return SyncHistoryDivideColumnResponse.success();
    }

    private int updateTables(List<DividingColumnInfo> dividingColumnInfoList, String idColumn, String id) {
        Map<String, List<DividingColumnInfo>> groupedUpdateInfos = dividingColumnInfoList.stream()
            .collect(Collectors.groupingBy(DividingColumnInfo::getDivideTable));
        int count = 0;
        for (Map.Entry<String, List<DividingColumnInfo>> entry : groupedUpdateInfos.entrySet()) {
            String tableName = entry.getKey();
            List<DividingColumnInfo> updateInfos = entry.getValue();

            StringBuilder sqlBuilder = new StringBuilder("UPDATE " + tableName + " SET ");

            List<Object> params = new ArrayList<>();

            for (int i = 0; i < updateInfos.size(); i++) {
                DividingColumnInfo dividingColumnInfo = updateInfos.get(i);
                String columnName = dividingColumnInfo.getColumnName();
                String content = dividingColumnInfo.getContent();

                sqlBuilder.append(columnName).append(" = ?");

                params.add(content);

                if (i < updateInfos.size() - 1) {
                    sqlBuilder.append(", ");
                }
            }
            sqlBuilder.append(" WHERE ").append(idColumn).append(" = ?");
            params.add(id);

            String sql = sqlBuilder.toString();

            count += jdbcTemplate.update(sql, params.toArray());
        }
        return count;
    }

    @Data
    public static class SyncHistoryDivideColumnResponse<T> {
        /**
         * 错误码, 0 代表成功
         */
        private int code;
        /**
         * 错误信息
         */
        private String message;
        /**
         * 分词列更新失败的data
         */
        private List<T> missingDataList;

        public SyncHistoryDivideColumnResponse(int code, String message, List<T> missingDataList) {
            this.code = code;
            this.message = message;
            this.missingDataList = missingDataList;
        }

        public static SyncHistoryDivideColumnResponse<?> success() {
            return new SyncHistoryDivideColumnResponse<>(0, "success", null);
        }

        public static <T> SyncHistoryDivideColumnResponse<T> fail(List<T> missingDataList, String message) {
            return new SyncHistoryDivideColumnResponse<>(1, message, missingDataList);
        }
    }

    @Data
    public static class SyncHistoryDivideColumnRequest<T> {
        /**
         * 实体类
         */
        private Class<T> entityClazz;
        /**
         * 表名
         */
        private String tableName;
        /**
         * id列名
         */
        private String idColumnName;
        /**
         * id的Getter方法
         */
        private Function<T, String> idGetter;
        /**
         * 租户id
         */
        private String tenantId;
        /**
         * 原字段的解密function
         */
        private Function<String, String> decryptFunction;
    }
}
